package com.dtwave.cfstreaming.process.function;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.cfstreaming.bean.ProcessData;
import com.dtwave.utils.JsonUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class ProcessMapFunction extends RichFlatMapFunction<JSONObject, ProcessData> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void flatMap(JSONObject jsonObject, Collector<ProcessData> collector) throws Exception {
        String name = JsonUtils.getString(jsonObject, true, "name");
        String age = JsonUtils.getString(jsonObject, true, "age");
        String source = JsonUtils.getString(jsonObject, true, "source");
        String sex = JsonUtils.getString(jsonObject, true, "sex");
        String password = JsonUtils.getString(jsonObject, true, "password");
        ProcessData processData = new ProcessData(
                name,
                age,
                password,
                sex,
                source
        );
        collector.collect(processData);
    }
}
